Thanks for the issue!
I think I can see other ways to do this without the need of making KafkaMetrics pubblic. But first, let me check if my understanding is correct, are you saying that:

1. In fs2-kafka there is no way to inject or get org.apache.kafka.clients.consumer.Consumer and org.apache.kafka.clients.producer.Producer even though fs2-kafka uses them under the hood?
2. There is a method in fs2-kafka that gives you back the metrics (Map<MetricName, Metric>) that you can pass to one of the ctors of KafkaMetrics? Is it fs2.kafka.consumer.KafkaMetrics#metrics?

I think #1 should be fixed in fs2-kafka, would you mind opening an issue there and link it here? (Unless they don't want to leak Kafka types on their public API but I'm not sure if that's a concern since fs2.kafka.consumer.KafkaMetrics is doing that).

For #2, the proper solution would be moving the instrumentation to the Java Kafka Client, we have an issue for this, would you mind commenting on it so that Kafka devs will see there is a user need? https://issues.apache.org/jira/browse/KAFKA-15191.

Other than these, I think there is a better solution to support other Kafka clients than making KafkaMetrics public: adding a public/protected overload to the already public classes that extend KafkaMetrics. The problem with this (and with making KafkaMetrics) public is that KafkaMetrics heavily depends on the behavior of the Java Kafka Client. Surfacing such an API might seem that KafkaMetrics should work with clients that are not depending on the Java Kafka Client. This might be ok though if we document this.

Hello @jonatan-ivanov, thank you for your response.
After your advice, I've added more constructors to the (already public) KafkaClientMetrics class, with the documentation stressing that the metrics need to come from genuine Java Kafka Client because it's being relied upon. That should make everybody happy.

Please let me know your thoughts.

Thanks for the changes, I think I might have not been clear about this. Before even considering changing anything in Micrometer, I would like to explore the first two (preferred) options I mentioned above.
You did not answer my questions above, I will assume the answer to both of them is yes so I created fs2-kafka/#1409, please feel free to comment on it (as well as KAFKA-15191).

Hi, I think it probably won't be possible to do that in fs2-kafka, but I may be wrong, it's worth asking. Let's see what we hear from the fs2-kafka developers.

Regarding KAFKA-15191, I'm not a Kafka developer, I don't have an account :D

Regarding KAFKA-15191, I'm not a Kafka developer, I don't have an account

Anyone can make an account. At the top of the page there should be a banner with the following message:

Public signup for this instance is disabled. Go to our Self serve sign up page to request an account.

The fs2-kafka people don't seem to be very responsive. Are you still open to merging this PR @jonatan-ivanov ?
It would solve the problem for us and hopefully for other people as well.

I pinged them on the issue, I rebased your PR, added an extra ctor (we added one in the meantime), some javadoc and polished it. I think we can merge it either in 1.16.0-M1 (released in July) or M2 (released in August).

So we have an answer from the fs2-kafka folks...

What they say is true, you can hijack the underlying Java Kafka object during the creation. So it's technically possible. But I think that's a very in-elegant solution. I think adding more constructors to KafkaClientMetrics, like this PR does, is cleaner and more flexible, if you @jonatan-ivanov wouldn't be against it 🙏

@jonatan-ivanov there are potential issues with concurrency when exposing the underlying clients from fs2-kafka. Are you OK with merging this PR?

What they say is true, you can hijack the underlying Java Kafka object during the creation. So it's technically possible. But I think that's a very in-elegant solution.

This is what dependency injection is all about unless I misunderstood something (my Scala fu is also ~10 years old).

@jonatan-ivanov there are potential issues with concurrency when exposing the underlying clients from fs2-kafka. Are you OK with merging this PR?

Could you please elaborate on this? The underlying kafka client must be thread-safe otherwise we can throw out the whole support for it from Micrometer (Micrometer also calls the kafka client from its own thread).

When we merge this PR, we'll be able to integrate Micrometer with fs2-kafka via the Supplier<Map<MetricName, ? extends Metric>> metricsSupplier. That will be implemented via the already existing fs2-kafka's KafkaMetrics[F[_]]#metrics: F[Map[MetricName, Metric]] which ensures thread safety.

What concurrency issues would we face if we would not merge this in but we would use MkConsumer and MkProducer to inject the client and also inject the same client to KafkaClientMetrics? Would not that be a working solution as of today?

What concurrency issues would we face if we would [...] use MkConsumer and MkProducer to inject the client and also inject the same client to KafkaClientMetrics

I'm not sure to be honest. But MkConsumer and MkProducer is awkward to use. And with KafkaMetrics[F[_]]#metrics: F[Map[MetricName, Metric]] we can be sure that there won't be any concurrency issues so that's why I think it's the most promising option.

with KafkaMetrics[F[_]]#metrics: F[Map[MetricName, Metric]] we can be sure that there won't be any concurrency issues

I'm not following. Why does that change anything compared to Micrometer receiving the client instance and calling the Kafka API for getting the metrics, as we do now in the constructors? Ultimately that KafkaMetrics is still calling the metrics API on the Kafka client, the same as we would, right? And that will happen at the same time as the client is being used by other threads, right? If there is a concurrency issue (I don't think there is), then it is an issue regardless.

I think adding more constructors to KafkaClientMetrics, like this PR does, is cleaner and more flexible

More flexible has a nice nuance to it, but from the perspective of maintenance, we're going to need to support 4 new public constructors in a class that already has a lot of constructors and we are exposing what previously was an implementation detail as public API with this change. We don't want to support arbitrary maps of metrics from unknown sources - we want to support metrics from the official Kafka clients. It happens to be that your use case (probably) is ultimately sourcing them from the official Kafka client, but that's not enforced by the API proposed in this PR.

At the end of the day, it probably won't be a big issue to add this, but it's not very enticing to open ourselves up to issues we don't have to deal with now. What does the alternative look like? How bad is it really to get the underlying client instance from the MkConsumer and MkProducer?

The implementation of KafkaMetrics[F[_]]#metrics: F[Map[MetricName, Metric]] is very special. It ensures that it calls the underlying Java Kafka client from the correct thread.

Indeed, you are correct that the new constructors in this PR don't enforce that the Metrics are coming from the Java Kafka client. But they give us more flexibility, which is something we'd be grateful for. Thank you 🙏

The implementation of KafkaMetrics[F[_]]#metrics: F[Map[MetricName, Metric]] is very special. It ensures that it calls the underlying Java Kafka client from the correct thread.

I can't understand why that would need to be the case. Can you point to some Kafka documentation explaining the concern for calling the Kafka client's metrics method from arbitrary threads? As @jonatan-ivanov previously shared, we do not call it from the same thread users will be using and have never heard of or are aware of any issue. If there is an issue, we'd like to know about it. If there isn't an issue, then the implementation in fs2-kafka probably doesn't need to be special.

I'm no Kafka expert, but the fs2-kafka maintainer @vlovgr has expressed concern about thread safety in this comment

• Make Consumer or Producer injectable or make it available for the users fd4s/fs2-kafka#1409 (comment)

Do you know if there are any potential issues with thread safety in this case? At least for the consumer, fs2-kafka uses a dedicated thread for interacting with the Java KafkaConsumer. If not, we could probably consider adding fs2.kafka.consumer.KafkaConsumerClient and fs2.kafka.producer.KafkaProducerClient traits (similar to the existing ones) to expose these. We should still document the general thread safety concerns here though.

I think the thread safety concerns of using the Kafka client for consuming messages from Kafka is different than retrieving metrics from the Kafka client, which is all Micrometer is concerned with doing.

Somewhat tangentially, I've noticed today that Kafka has an interface MetricsReporter, and I wonder if we couldn't better instrument using that, which would require a rework of the KafkaMetrics, and would be another reason we would not want to add the constructors in this PR. It would take some more investigation to see if MetricsReporter does what we need or not, though.